package org.benben.shortlink.project.config;

import lombok.RequiredArgsConstructor;
import org.benben.shortlink.project.mq.consumer.ShortLinkStatsSaveConsumer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Redis Stream 消息队列配置
 *
 * @author benben
 */
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    /**
     * Redis 连接工厂类
     */
    private final RedisConnectionFactory redisConnectionFactory;

    /**
     * 消息队列消费者
     */
    public final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;

    /**
     * Stream Key
     */
    @Value("${spring.data.redis.channel-topic.short-link-stats}")
    private String topic;

    /**
     * 消费者组
     */
    @Value("${spring.data.redis.channel-topic.short-link-stats-group}")
    private String group;

    /**
     * 创建执行消息的线程池
     *
     * @return
     */
    @Bean
    public ExecutorService asyncStreamConsumer() {
        AtomicInteger index = new AtomicInteger();
        int processors = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
                processors,
                processors + processors >> 1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                runnable -> {
                    Thread thread = new Thread(runnable);
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }

    /**
     * 创建监听器
     *
     * @param asyncStreamConsumer 消费者
     * @return
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                //一次最多获取多少条消息
                .batchSize(10)
                //执行从 Stream 拉取到消息的任务流程
                .executor(asyncStreamConsumer)
                //如果没有拉取到消息，需要阻塞的时间。不能大于 ${spring.data.redis.timeout}，否则会超时
                .pollTimeout(Duration.ofSeconds(3))
                .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer = StreamMessageListenerContainer.create(redisConnectionFactory, options);
        streamMessageListenerContainer.receiveAutoAck(Consumer.from(group, "stats-consumer"), StreamOffset.create(topic, ReadOffset.lastConsumed()), shortLinkStatsSaveConsumer);
        return streamMessageListenerContainer;
    }

}
